package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo16Cache {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()

    conf.setAppName("cache")
    conf.setMaster("local")

    val sc = new SparkContext(conf)

    //设置checkpoint的路径
    //sc.setCheckpointDir("data/checkpoint")

    val linesRDD: RDD[String] = sc.textFile("data/students.txt")


    //切分数据
    val studentRDD: RDD[(String, String, Int, String, String)] = linesRDD.map((line: String) => {
      println("map")
      val split: Array[String] = line.split(",")
      val id: String = split(0)
      val name: String = split(1)
      val age: Int = split(2).toInt
      val gender: String = split(3)
      val clazz: String = split(4)
      (id, name, age, gender, clazz)
    })

    /**
     * 当同一个rdd被多次使用 时将可以rdd缓存起来，只需要计算一次
     *
     * cache的缓存级别时MEMORY_ONLY（内存）
     *
     * MEMORY_ONLY： 只将数据房子内存中
     * MEMORY_AND_DISK_SER: 内存放不下再放磁盘，同时会对数据做序列化（将一个分区的数据序列化成一个字节数组）
     * 缓存的数据时缓存在Executor的内存和磁盘上的，任务执行完成会自动清理
     */
    //studentRDD.cache()

    //指定缓存级别
    studentRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)


    /**
     * checkpoint: 可以将rdd的数据缓存到hdfs中
     * 需要设置checkpoint的路径
     * checkpoint主要用在 spark streaming
     */
    //studentRDD.checkpoint()


    //统计班级的人数
    val clazzRDD: RDD[(String, Int)] = studentRDD.map {
      case (_: String, _: String, _: Int, _: String, clazz: String) => (clazz, 1)
    }
    val clazzNumRDD: RDD[(String, Int)] = clazzRDD.reduceByKey(_ + _)
    clazzNumRDD.foreach(println)

    //统计性别的人数
    val genderRDD: RDD[(String, Int)] = studentRDD.map {
      case (_: String, _: String, _: Int, gender: String, _: String) => (gender, 1)
    }
    val genderNumRDD: RDD[(String, Int)] = genderRDD.reduceByKey(_ + _)
    genderNumRDD.foreach(println)

  }

}
